/**
 * 
 */

package com.wutianyi.queue.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.lang.RandomStringUtils;

import com.wutianyi.pool.concurrent.CountTime;
import com.wutianyi.queue.Task;
import com.wutianyi.queue.TaskObject;
import com.wutianyi.queue.TaskQueue;

/**
 * @author wutianyi
 * 
 */
public class TaskQueueImpl<T extends Task> implements TaskQueue<T> {

	/**
	 * 最大的segment数
	 */
	private static int						MAX_SEGMENT_SIZE		= 5;
	private static int						MAX_EACH_SEGMENT_SIZE	= 100000;

	private int								segmentSize				= 0;
	private int								eachSegmentSize			= 0;

	private Segment<T>[]					segments;
	/**
	 * 保存超过segment的task
	 * 由用户自己设置，如果没有设置，那么超过的task将丢弃
	 */
	private TaskQueue<T>					outOfSizeQueue;

	/**
	 * outOfSizeQueue的索引
	 */
	private int								outOfSizeQueueIndex		= 0;

	private ReentrantLock					locks					= new ReentrantLock();

	private LinkedBlockingQueue<Integer>	outputIndex				=
																			new LinkedBlockingQueue<Integer>();

	public TaskQueueImpl(int segmentSize, int eachSegmentSize) {

		this.segmentSize = segmentSize;
		this.outOfSizeQueueIndex = segmentSize;

		segments = new Segment[segmentSize];
		for(int i = 0; i < segmentSize; i++) {
			segments[i] = new Segment<T>();
		}
	}

	public TaskQueueImpl() {

		this(MAX_SEGMENT_SIZE, MAX_EACH_SEGMENT_SIZE);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#drainTo(java.util.Collection, int)
	 */
	@Override
	public int drainTo(Collection<T> c, int size) {

		return 0;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#offer(java.lang.Object)
	 */
	@Override
	public boolean offer(T t) {

		int index = Math.abs(t.getTaskId().hashCode()) % segmentSize;

		if(segments[index].size() > eachSegmentSize) {
			index = this.outOfSizeQueueIndex;

			if(null != outOfSizeQueue) {
				outputIndex.offer(index);
				return outOfSizeQueue.offer(t);
			}
		}
		outputIndex.offer(index);
		return segments[index].offer(t);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#poll()
	 */
	@Override
	public T poll() {

		if(outputIndex.size() == 0) {
			return null;
		}

		Integer index = outputIndex.poll();
		if(null == index) {
			return null;
		}

		if(index == outOfSizeQueueIndex) {
			if(null != outOfSizeQueue) {
				return outOfSizeQueue.poll();
			}
		}
		return segments[index].poll();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#size()
	 */
	@Override
	public int size() {

		int size = 0;
		try {
			locks.lock();

			for(Segment<T> segment : segments) {
				size += segment.size();
			}
			if(null != outOfSizeQueue) {
				size += outOfSizeQueue.size();
			}
		}finally {
			locks.unlock();
		}
		return size;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#drainTo(int)
	 */
	@Override
	public List<T> drainTo(int size) {

		if(0 == size || 0 == outputIndex.size()) {
			return null;
		}
		List<T> tasks = new ArrayList<T>();

		for(int i = 0; i < size; i++) {
			T t = poll();
			if(null == t) {
				break;
			}
			tasks.add(t);
		}

		return tasks;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#batchAddTask(java.util.List)
	 */
	@Override
	public void batchAddTask(List<T> tasks) {

		if(null == tasks || 0 == tasks.size()) {
			return;
		}
		for(T t : tasks) {
			offer(t);
		}
	}

	public int outputIndexSize() {

		return outputIndex.size();
	}

	public TaskQueue<T> getOutOfSizeQueue() {

		return outOfSizeQueue;
	}

	public void setOutOfSizeQueue(TaskQueue<T> outOfSizeQueue) {

		this.outOfSizeQueue = outOfSizeQueue;
	}

	public int getEachSegmentSize() {

		return eachSegmentSize;
	}

	public static void main(String[] args) throws InterruptedException {

		final TaskQueueImpl<TaskObject> queue = new TaskQueueImpl<TaskObject>();
		final AtomicInteger count = new AtomicInteger(0);
		CountTime countTime = new CountTime(new Runnable() {

			@Override
			public void run() {

				for(int i = 0; i < 10000; i++) {
					TaskObject obj =
							new TaskObject(RandomStringUtils.random(20), Thread.currentThread()
									.getName(), new HashMap<String, String>());
					queue.offer(obj);
				}
			}
		}, 10);
		System.out.println(queue.size());
		System.out.println(queue.outputIndexSize());
		System.out.println("finish producer!");
		Thread.sleep(2000);
		System.out.println("Begin consumer");
		CountTime consumer = new CountTime(new Runnable() {

			@Override
			public void run() {

				boolean isRun = true;
				int consumerCount = 0;
				while(isRun) {
					TaskObject obj = queue.poll();
					if(null != obj) {
						consumerCount++;
						count.getAndIncrement();
					}else {
						isRun = false;
						System.out.println("Thread: " + Thread.currentThread().getName()
								+ " finish " + consumerCount + " task !");
					}
				}
			}
		}, 10);
		System.out.println(queue.outputIndexSize());
		System.out.println(queue.size());
		System.out.println(count.get());
		System.out.println("consumer Finish!");
		System.out.println("Begin batch Add Tasks!");
		Thread.sleep(2000);
		CountTime batchAdd = new CountTime(new Runnable() {

			@Override
			public void run() {

				for(int i = 0; i < 100; i++) {
					List<TaskObject> tasks = new ArrayList<TaskObject>();
					for(int j = 0; j < 100; j++) {
						TaskObject obj =
								new TaskObject(RandomStringUtils.random(20), Thread.currentThread()
										.getName(), new HashMap<String, String>());
						tasks.add(obj);
					}
					queue.batchAddTask(tasks);
				}
			}
		}, 10);
		System.out.println(queue.size());
		System.out.println(queue.outputIndexSize());
		System.out.println("finish batch add Tasks !");
		Thread.sleep(2000);
		System.out.println("Begin batch remove task!");
		count.set(0);
		CountTime batchRemove = new CountTime(new Runnable() {

			@Override
			public void run() {

				boolean isRun = true;
				int c = 0;
				while(isRun) {
					List<TaskObject> tasks = queue.drainTo(10);
					if(null != tasks && 0 != tasks.size()) {
						c += tasks.size();
						count.getAndAdd(tasks.size());
					}else {
						isRun = false;
						System.out.println("Thread : " + Thread.currentThread().getName()
								+ " finish " + c + " tasks!");
					}
				}

			}
		}, 10);
		System.out.println(queue.outputIndexSize());
		System.out.println(queue.size());
		System.out.println(count.get());
	}

}
